package com.idroc.coelacanth.core.membership;

import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.SortedMap;
import java.util.TreeMap;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;

import org.jboss.netty.channel.ChannelFuture;
import org.jboss.netty.channel.ChannelFutureListener;
import org.jboss.netty.channel.ChannelHandlerContext;
import org.jboss.netty.channel.MessageEvent;
import org.jboss.netty.channel.SimpleChannelUpstreamHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.idroc.coelacanth.core.message.Message;
import com.idroc.coelacanth.core.message.MessageDispatchHandler;
import com.idroc.coelacanth.service.MessageService;
import com.idroc.coelacanth.service.ThreadPoolService;
import com.idroc.coelacanth.util.RandomUtil;

/**
 * 流言协议
 * @author yuyoo (yuyoo4j@163.com)
 * @date 2009-12-2 上午11:22:30
 * 端点之间通过流言协议进行传播成员状态
 */
public class Gossiper {
	
	private static final Logger log = LoggerFactory.getLogger(Gossiper.class);
	
	/**
	 * 建立gossip的端点
	 */
	private Endpoint endpoint = null;
	
	/**
	 * 初始化引用信息端点
	 */
	private Endpoint quoteEndpoint = null; 
	
	/**
	 * 端点元数据
	 */
	private EndpointMeta currentMeta = null;
	 	
	/**
	 * 加入状态端点集合
	 */
	private final SortedMap<Integer, EndpointMeta> joinMetas = new TreeMap<Integer, EndpointMeta>();
	 	
	/**
	 * 开放状态端点集合
	 */
	private final SortedMap<Integer, EndpointMeta> openMetas = new TreeMap<Integer, EndpointMeta>();
	 	
	/**
	 * 关闭状态端点集合
	 */
	private final SortedMap<Integer, EndpointMeta> closeMetas = new TreeMap<Integer, EndpointMeta>();
	 		
	/**
	 * 状态观察者集合
	 */
	private final List<MemberStateObserver> observers = new ArrayList<MemberStateObserver>();
	 	
	/**
	 * 消息服务
	 */
	private MessageService msgService = null;
	
	private final long heartbeatInterval = 5000; // 12s
	private final long probeOpenInterval = 120000; // 120s
	private final long heartbeatLifecycle = 30000; // 120s
	private final long pulseVersionBeginning = 0; // 0
	
	private boolean haveToRandomSend = false;
	
	public Gossiper(MessageService ms, Endpoint s, Endpoint q) {
				 
		msgService = ms;
		endpoint = s;
		quoteEndpoint = q; 
		addObserver(new GossipSelfMaintainManager()); // 自维护观察者
	}	
			
	public Gossiper(MessageService ms, Endpoint s) {
		
		this(ms, s, null);
	}
	
	/**
	 * 启动流言协议
	 */
	public void start() {
		  		
		log.debug("流言协议Gossip启动");
		haveToRandomSend = false;
		registerMessageHandler();
		startGossipTask();
		selfStartAction();
	}
	
	/**
	 * 启动行为
	 */
	private void selfStartAction() {

		// 启动行为
		final Runnable startAction = new Runnable() {
			public void run() { 				
				currentMeta = memberJoinAction(endpoint); // 加入端点集群
				log.debug("启动行为将睡眠一个心跳时间间隔[{}ms], 保证'加入集群消息'在集群传播完成,保证每个开放端点都接收到消息", heartbeatInterval);
				// 睡眠一个心跳时间间隔, 保证'加入集群消息'在集群传播完成
				try { Thread.sleep(heartbeatInterval); } catch (Exception ex) { /* 消除异常 */ }
				memberJoinToOpenAction(endpoint); // 对端点集群开放服务				
			}			
		};
		if (null == quoteEndpoint) { // 自启动
			log.debug("不存在'引用端点', 自启动.");
			// startAction.run();
			ThreadPoolService.getMaintainThreadPool().submit(startAction);	
		} else { // 通过"引用端点"获取成员集合启动 
			log.debug("通过'引用端点'获取成员集合启动.");
			MessageDispatchHandler dispatcher = msgService.getServerMessageFactory().getMessageDispatcher();
			dispatcher.addUpMessageHandler(MemberSetResponseMessage.class, new MemberSetResponseMessageHandler() {

				@Override
				public void messageReceived(ChannelHandlerContext ctx,
						MessageEvent e) throws Exception {
					
					log.debug("通过'引用端点'获取成员集合, 获取响应消息[{}]", e.getMessage());
					super.messageReceived(ctx, e); // 处理消息				
					ThreadPoolService.getMaintainThreadPool().submit(startAction);					
				}
				
			});
			buildCurrentMemberMeta(quoteEndpoint); // 建立当前集群的端点元数据. 发送端点成员集合请求消息给引用端点			
		}
	}
	
	/**
	 * 获取并处理当前端点集群信息
	 * @param t -- 引用端点
	 */
	private ChannelFuture buildCurrentMemberMeta(Endpoint quote) {
		
		MemberSetRequestMessage msg = new MemberSetRequestMessage();
		msg.setFrom(endpoint);
		int rc = MemberState.OPEN.getCode() | MemberState.CLOSE.getCode() | MemberState.JOIN.getCode();
		msg.setRequestMemberStateCode(rc);
		return sendMessage(quote, msg);
	}
	
	/**
	 * 注册消息处理器
	 */
	private void registerMessageHandler() {
		 
		MessageDispatchHandler dispatcher = msgService.getServerMessageFactory().getMessageDispatcher();
		// 处理"心跳消息"
		dispatcher.addUpMessageHandler(HeartBeatBroadcastMessage.class, new HeartBeatBroadcastMessageHandler());
		// 处理"心跳失效请求消息"
		dispatcher.addUpMessageHandler(HeartbeatInvalidRequestMessage.class, new HeartbeatInvalidRequestMessageHandler());
		// 处理"心跳失效响应消息"
		dispatcher.addUpMessageHandler(HeartbeatInvalidResponseMessage.class, new HeartbeatInvalidResponseMessageHandler());
		// 处理"端点从开放状态转入关闭状态消息"
		dispatcher.addUpMessageHandler(MemberOpenToCloseMessage.class, new MemberOpenToCloseMessageHandler());
		// 处理"端点从关闭状态转入开放状态消息"
		dispatcher.addUpMessageHandler(MemberCloseToOpenMessage.class, new MemberCloseToOpenMessageHandler());
		// 处理"端点成员加入消息"
		dispatcher.addUpMessageHandler(MemberJoinMessage.class, new MemberJoinMessageHandler());
		// 处理"端点从加入状态转入开放状态消息"
		dispatcher.addUpMessageHandler(MemberJoinToOpenMessage.class, new MemberJoinToOpenMessageHandler());
		// 处理"端点成员请求消息"
		dispatcher.addUpMessageHandler(MemberSetRequestMessage.class, new MemberSetRequestMessageHandler());
		// 处理"端点成员响应消息"
		dispatcher.addUpMessageHandler(MemberSetResponseMessage.class, new MemberSetResponseMessageHandler());
		 
	}
	
	/**
	 * 启动流言协议任务
	 */
	private void startGossipTask() {
		 
		/**
		 * 流言协议:心跳任务
		 */
		Runnable heartbeatTask = new Runnable() {

			public void run() {
				
				long current = System.currentTimeMillis();
				log.debug("开始心跳检查, 任务号[{}], 心跳检查间隔[{}ms]", current, heartbeatInterval);
				try {	
					// 当前节点处于open状态
					if (null == currentMeta) {
						log.debug("端点[{}]当前端点没有元数据. 放弃当前心跳任务", endpoint);
						return;
					}
					
					if (currentMeta.memberState == MemberState.OPEN) {
						// 下一个心跳状态
						long nextPV = currentMeta.heartbeatState.getPulseVersion() + 1;
						long nextPT = current; 
						currentMeta.heartbeatState.setPulseVersion(nextPV);
						currentMeta.heartbeatState.setPulseTime(nextPT);
						currentMeta.heartbeatState.setApperceiveTime(nextPT);
						currentMeta.heartbeatState.setLifecycle(heartbeatLifecycle);
						// 下一个心跳消息
						HeartBeatBroadcastMessage msg = new HeartBeatBroadcastMessage();
						msg.setHopToLive(openMetas.size());
						msg.setLifecycle(heartbeatLifecycle);
						msg.setPulseTime(nextPT);
						msg.setPulseVersion(nextPV);
						msg.setNode(currentMeta.endpoint);
						
						gossipToNext(msg);
					}
				} catch (Exception ex) {
					log.error("开始心跳检查异常", ex);
				}
				ThreadPoolService.getMaintainSchduledThreadPool().schedule(this, heartbeatInterval, TimeUnit.MILLISECONDS);
				log.debug("完成心跳检查, 任务号[{}]", current);
			}			
		};
		ThreadPoolService.getMaintainSchduledThreadPool().schedule(heartbeatTask, heartbeatInterval, TimeUnit.MILLISECONDS);
		
		/**
		 * 检查心跳状态有效性任务
		 */
		Runnable probeOpenStateTask = new Runnable() {

			public void run() {
				
				long current = System.currentTimeMillis();
				log.debug("开始开放状态端点集合自检查,任务号[{}], 开放状态端点集合(openMetas)[{}]", current, openMetas); 
				try {					
					for (Integer key : openMetas.keySet()) {
						
						EndpointMeta item = openMetas.get(key);
						log.debug("检查端点[{}]", item.endpoint);
						try {				
							// 处于开发状态
							if (item.memberState == MemberState.OPEN) {
								
								log.debug("端点处于开放状态");
								long at = item.heartbeatState.getApperceiveTime();
								long lc = item.heartbeatState.getLifecycle();
								if (at + lc < current) {
									HeartbeatInvalidRequestMessage msg = new HeartbeatInvalidRequestMessage();
									msg.setFrom(currentMeta.endpoint);
									msg.setPulseVersion(item.heartbeatState.getPulseVersion());
									msg.setPulseTime(item.heartbeatState.getPulseTime());
									msg.setLifecycle(item.heartbeatState.getLifecycle());
									item.memberState = MemberState.CLOSE;
									sendMessage(item.endpoint, msg);
								}
							} else if (item.memberState == MemberState.CLOSE) { // 处于关闭状态, 曾发生过心跳失效, 发送过心跳失效请求消息
								
								log.debug("端点处于关闭状态,曾发生过心跳失效");
								memberOpenToCloseAction(item.endpoint);
							}
						} catch (Exception iex) {
							log.error("检查异常, 开放状态端点元数据:{}", item);
						}
					}
				} catch (Exception ex) {
					log.error("检查失效节点异常", ex);
				}
				ThreadPoolService.getMaintainSchduledThreadPool().schedule(this, probeOpenInterval, TimeUnit.MILLISECONDS);
				log.debug("完成开放状态端点集合自检查任务[{}]", current);
			}
		};
		ThreadPoolService.getMaintainSchduledThreadPool().schedule(probeOpenStateTask, probeOpenInterval, TimeUnit.MILLISECONDS);
	}
	
	/**
	 * 将一个端点关闭
	 * @param ep -- 端点
	 */
	public void memberOpenToCloseAction(Endpoint ep) {
		
		log.debug("关闭端点[{}]", ep);
		
		Integer key = Integer.valueOf(ep.hashCode());
		final EndpointMeta item = openMetas.remove(key);
		if (null != item) {
			
			item.memberState = MemberState.CLOSE; 
			closeMetas.put(key, item);
			// 通知 观察者
			noticeMemberStateObserver(MemberState.CLOSE, item); 
			
			// 流言 端点成员关闭消息
			MemberOpenToCloseMessage msg = new MemberOpenToCloseMessage();
			msg.setNode(item.endpoint);
			msg.setPulseVersion(item.heartbeatState.getPulseVersion());
			msg.setPulseTime(item.heartbeatState.getPulseTime());
			msg.setLifecycle(item.heartbeatState.getLifecycle());
			msg.setHopToLive(openMetas.size());
			gossipToNextTask(msg);
		} else {
			log.debug("在开放状态集合未找到端点[{}]", ep);
		}
	}
	
	/**
	 * 端点状态发生改变 通知 端点成员状态观察者
	 * @param state -- 端点成员状态
	 * @param em -- 发生状态改变的端点元数据
	 */
	private void noticeMemberStateObserver(final MemberState state, final EndpointMeta em) {
		
		for (MemberStateObserver io : observers) {
			try {
				final MemberStateObserver fio = io;
				Runnable task = new Runnable() {
					public void run() {
						fio.update(state, em);						
					}
				};
				ThreadPoolService.getMaintainThreadPool().submit(task);
			} catch (Exception ioex) {
				log.error("端点[{}]成员状态[{}]发生改变", em, state);
				log.error("通知成员状态观察者[{}]异常", io);
				log.error("通知异常", ioex);
			}
		}
	}
	
	/**
	 * 使用策略将消息传播
	 * @param msg -- 消息 
	 */ 
	public void gossipToNext(final Message msg) {
		
		log.debug("流言协议(gossip)传播消息[{}]", msg);
		
		// 找出下一个节点 
		
		Endpoint currentEndpoint = (null != currentMeta) ? currentMeta.endpoint : endpoint;
 
		SortedMap<Integer, EndpointMeta> tails = openMetas.tailMap(Integer.valueOf(currentEndpoint.hashCode() + 1));
		EndpointMeta next = null;
		if (tails.size() > 0) { 
			
			Iterator<Integer> it = tails.keySet().iterator();
			EndpointMeta tmp = null;
			while (it.hasNext()) {
				tmp = openMetas.get(it.next());
				if (tmp.memberState == MemberState.OPEN) {
					next = tmp;
					break;
				}
			}						
		} else if (openMetas.size() > 0) { 

			Iterator<Integer> it = openMetas.keySet().iterator();
			EndpointMeta tmp = null;
			while (it.hasNext()) {
				tmp = openMetas.get(it.next());
				if (tmp.memberState == MemberState.OPEN) {
					next = tmp;
					break;
				}
			}
		}

		// 如果下个节点不是自己(如果是自己,说明整个集群中只有一个节点)
		if (null != next && !currentEndpoint.equals(next.endpoint)) {	
			// 发布心跳给下个节点
			log.debug("向下一个端点[{}]发送消息", next.endpoint);
			sendMessage(next.endpoint, msg); 
			
			// 随机发送给一个节点
			if (openMetas.size() > 1 && haveToRandomSend) {
				
				Integer randomKey = (Integer)RandomUtil.randomGetKey(openMetas);
				EndpointMeta random = openMetas.get(randomKey);
				if (!currentEndpoint.equals(random.endpoint)) {
					log.debug("向随机端点[{}]发送消息", random.endpoint);
					sendMessage(random.endpoint, msg);
				}
			}
		} else { 
			log.debug("端点集群中只有当前端点[{}].", currentEndpoint);  
		} 
	} 
	
	/**
	 * 提交[使用策略将消息传播]任务到线程池执行
	 * @param msg -- 消息
	 * @return  -- 任务结果
	 */
	@SuppressWarnings("unchecked")
	public Future gossipToNextTask(final Message msg) {
		
		Runnable task = new Runnable() {

			public void run() {
				gossipToNext(msg);
			}			
		};
		return ThreadPoolService.getMaintainThreadPool().submit(task);
	}
	 	
	/**
	 * 向端点发送消息
	 * @param ep -- 端点
	 * @param msg -- 消息
	 * @return -- 通道结果(ChannelFuture)
	 */
	public ChannelFuture sendMessage(final Endpoint ep, final Message msg) {
		
		
		ChannelFuture cf = msgService.sendMessage(ep, msg);
		cf.addListener(new ChannelFutureListener() {
	
				public void operationComplete(ChannelFuture writeFuture)
						throws Exception { 
	
					if (!writeFuture.isSuccess()) {
						log.error("向端点[{}]发送消息[{}]异常, 端点将被由开放转到关闭端点状态", ep, msg);
						memberOpenToCloseAction(ep);
					} 
				} 
		}); 
		return cf;
	}
	
	/**
	 * 提交[向端点发送消息]任务到线程池执行
	 * @param ep -- 端点
	 * @param msg -- 消息
	 * @return -- 任务结果
	 */
	@SuppressWarnings("unchecked")
	public Future sendMessageTask(final Endpoint ep, final Message msg) {
		
		Runnable task = new Runnable() {

			public void run() {
				sendMessage(ep, msg);
			}			
		};
		return ThreadPoolService.getMaintainThreadPool().submit(task);
	}
	
	/**
	 * 更新当前端点到加入状态集合
	 * @param ep -- 端点
	 * @return -- 端点元数据(EndpointMeta)
	 */
	public EndpointMeta memberJoinAction(Endpoint ep) {
		
		log.debug("更新当前端点到加入状态集合行为");
		
		final EndpointMeta em = new EndpointMeta();
		em.endpoint = ep;
		em.memberState = MemberState.JOIN;
		em.heartbeatState = new HeartBeatState();

		em.heartbeatState.setPulseVersion(pulseVersionBeginning);
		long current = System.currentTimeMillis();
		em.heartbeatState.setPulseTime(current);
		em.heartbeatState.setLifecycle(heartbeatLifecycle);
		em.heartbeatState.setApperceiveTime(current);
		
		Integer key = Integer.valueOf(ep.hashCode());
		joinMetas.put(key, em);
		// 通知 观察者
		noticeMemberStateObserver(MemberState.JOIN, em);  
		
		MemberJoinMessage msg = new MemberJoinMessage();
		msg.setNode(em.endpoint);
		msg.setPulseVersion(em.heartbeatState.getPulseVersion());
		msg.setPulseTime(em.heartbeatState.getPulseTime());
		msg.setLifecycle(em.heartbeatState.getLifecycle()); 
		msg.setHopToLive(openMetas.size());
		gossipToNext(msg); 
		return em;
	}
	
	/**
	 * 更新端点从加入状态集合到开放状态集合
	 * @param ep -- 端点
	 */
	public void memberJoinToOpenAction(Endpoint ep) {
		
		Integer key = Integer.valueOf(ep.hashCode());
		final EndpointMeta item = joinMetas.get(key);
		if (null != item) {
			
			long current = System.currentTimeMillis();
			item.heartbeatState.setPulseVersion(item.heartbeatState.getPulseVersion() + 1);
			item.heartbeatState.setPulseTime(current);
			item.heartbeatState.setApperceiveTime(current);
			item.heartbeatState.setLifecycle(heartbeatLifecycle);
			item.memberState = MemberState.OPEN;
			joinMetas.remove(key);
			openMetas.put(key, item);
			
			log.debug("更新当前端点到开放状态集合, 当前端点[{}]", ep);
			// 通知 观察者
			noticeMemberStateObserver(MemberState.OPEN, item);  
			
			MemberJoinToOpenMessage msg = new MemberJoinToOpenMessage();
			msg.setNode(ep);
			msg.setPulseVersion(item.heartbeatState.getPulseVersion());
			msg.setPulseTime(item.heartbeatState.getPulseTime());
			msg.setLifecycle(item.heartbeatState.getLifecycle());
			msg.setHopToLive(openMetas.size());
			gossipToNextTask(msg);
		} else {
			String errorMsg = "在加入状态集合中未找到端点, endpoint:" + ep;
			log.error(errorMsg);
			throw new RuntimeException(errorMsg);
		}
	}
	
	/**
	 * 更新端点到开放状态集合
	 * @param ep -- 端点
	 */
	public void memberCloseToOpenAction(Endpoint ep) {
		 
		Integer key = Integer.valueOf(ep.hashCode());
		final EndpointMeta item = closeMetas.remove(key);
		if (null != item) {
			item.memberState = MemberState.OPEN;
			long current = System.currentTimeMillis();
			item.heartbeatState.setPulseVersion(item.heartbeatState.getPulseVersion() + 1);
			item.heartbeatState.setPulseTime(current);
			item.heartbeatState.setApperceiveTime(current);
			openMetas.put(key, item);
			
			log.debug("更新端点到开放状态集合. Endpoint:{}", ep);
			// 通知 观察者
			noticeMemberStateObserver(MemberState.OPEN, item); 
			
			MemberCloseToOpenMessage msg = new MemberCloseToOpenMessage();
			msg.setNode(item.endpoint);
			msg.setPulseVersion(item.heartbeatState.getPulseVersion());
			msg.setPulseTime(item.heartbeatState.getPulseTime());
			msg.setLifecycle(item.heartbeatState.getLifecycle());
			msg.setHopToLive(openMetas.size());
			gossipToNextTask(msg);
		} else {
			String errorMsg = "当前端点状态未处于关闭状态集合";
			log.error(errorMsg);
			throw new RuntimeException(errorMsg);
		}
	} 
	
	/**
	 * 关闭Gossip
	 */
	public void stop() {
		
		log.debug("关闭Gossip"); 
	}
	
	/**
	 * 增加 端点成员状态观察者
	 * @param o -- 端点成员状态观察者
	 */
	public void addObserver(MemberStateObserver o) {
		
		log.debug("增加端点状态观察者[{}]", o);
		observers.add(o);
	}
	
	/**
	 * 移除 端点成员状态观察者
	 * @param o -- 端点成员状态观察者
	 */
	public void removeObserver(MemberStateObserver o) {
		
		log.debug("移除端点状态观察者[{}]", o);
		observers.remove(o);
	} 
	
	/**
	 * 获取消息服务
	 * @return -- 消息服务
	 */
	public MessageService getMsgService() {
		return msgService;
	}

	/**
	 * 获取 建立gossip的端点
	 * @return -- 端点
	 */
	public Endpoint getEndpoint() {
		return endpoint;
	} 

	/**
	 * 获取 当前端点元数据
	 * @return -- 端点元数据
	 */
	public EndpointMeta getCurrentMeta() {
		return currentMeta;
	}

	/**
	 * 设置当前端点元数据
	 * @param currentMeta -- 端点元数据 
	 */
	public void setCurrentMeta(EndpointMeta currentMeta) {
		this.currentMeta = currentMeta;
	}

	/**
	 * 获取 加入状态端点元数据集合
	 * @return -- 加入状态端点元数据集合
	 */
	public SortedMap<Integer, EndpointMeta> getJoinMetas() {
		return joinMetas;
	}

	/**
	 * 获取 开放状态端点元数据集合
	 * @return -- 开放状态端点元数据集合
	 */
	public SortedMap<Integer, EndpointMeta> getOpenMetas() {
		return openMetas;
	}

	/**
	 * 获取 关闭状态端点元数据集合
	 * @return -- 关闭状态端点元数据集合
	 */
	public SortedMap<Integer, EndpointMeta> getCloseMetas() {
		return closeMetas;
	}

	/**
	 * 获取 端点成员状态观察者
	 * @return -- 端点成员状态观察者List
	 */
	public List<MemberStateObserver> getObservers() {
		return observers;
	}
	
	/**
	 * Gossip自维护管理器
	 * @author yuyoo (yuyoo4j@163.com)
	 * @date 2009-12-8 下午02:33:03
	 */
	class GossipSelfMaintainManager implements MemberStateObserver {
		
		public void update(MemberState state, EndpointMeta item) {
			
			if (state == MemberState.CLOSE) {
				log.debug("端点[{}]由开放转关闭端点状态, 将关闭端点通道", item.endpoint);
				msgService.getEndpointConnectionManager().closeConnection(item.endpoint);
			}
		} 		
	}
	
	/**
	 * 心跳消息处理器
	 * @author yuyoo (yuyoo4j@163.com)
	 * @date 2009-12-4 下午05:13:18
	 */
	class HeartBeatBroadcastMessageHandler extends SimpleChannelUpstreamHandler {

		@Override
		public void messageReceived(ChannelHandlerContext ctx, MessageEvent e)
				throws Exception {
			
			HeartBeatBroadcastMessage msg = (HeartBeatBroadcastMessage)e.getMessage(); 
			log.debug("心跳消息处理器,消息:{}", msg);
			
			Integer key = Integer.valueOf(msg.getNode().hashCode());
			EndpointMeta item = openMetas.get(key);
			
			if (null != item && item.heartbeatState.getPulseVersion() < msg.getPulseVersion()
					&& item.heartbeatState.getPulseTime() < msg.getPulseTime()) {
				
				// 更新成员的心跳状态
				item.heartbeatState.setPulseVersion(msg.getPulseVersion());
				item.heartbeatState.setPulseTime(msg.getPulseTime());
				item.heartbeatState.setLifecycle(msg.getLifecycle());
				item.heartbeatState.setApperceiveTime(System.currentTimeMillis());
				
				// 消息跳转次数大于0, 需要被转发
				if (msg.getHopToLive() > 0) { 
					msg.setHopToLive(msg.getHopToLive() - 1);
					gossipToNext(msg);
				}
			} else  if (msg.getNode().equals(currentMeta.endpoint)) {
				// 心跳消息回传到当前端点
				if (msg.getPulseVersion() == currentMeta.heartbeatState.getPulseVersion()
						&& msg.getPulseTime() == currentMeta.heartbeatState.getPulseTime()
						&& msg.getLifecycle() == currentMeta.heartbeatState.getLifecycle()) {
					log.debug("心跳消息回传到当前端点");
				} else {
					log.debug("心跳消息回传到当前端点, 发现回传消息异常");
				}
			} else {
				log.debug("无效心跳消息[{}]", msg);
			}
		}
	}
	
	/**
	 * 心跳失效查询消息处理器
	 * @author yuyoo (yuyoo4j@163.com)
	 * @date Dec 5, 2009 9:52:03 PM
	 */
	class HeartbeatInvalidRequestMessageHandler extends SimpleChannelUpstreamHandler {

		@Override
		public void messageReceived(ChannelHandlerContext ctx, MessageEvent e)
				throws Exception {

			HeartbeatInvalidRequestMessage msg = (HeartbeatInvalidRequestMessage)e.getMessage();
			log.debug("心跳失效查询消息处理器,消息:{}", msg);
			
			HeartbeatInvalidResponseMessage res = new HeartbeatInvalidResponseMessage();
			res.setFrom(currentMeta.endpoint);
			res.setPulseVersion(currentMeta.heartbeatState.getPulseVersion());
			res.setPulseTime(currentMeta.heartbeatState.getPulseTime());
			res.setLifecycle(currentMeta.heartbeatState.getLifecycle());
			sendMessage(msg.getFrom(), res);
		}
		
	}
	
	/**
	 * 消息失效响应消息处理器
	 * @author yuyoo (yuyoo4j@163.com)
	 * @date 2009-12-7 上午09:46:22
	 */
	class HeartbeatInvalidResponseMessageHandler extends SimpleChannelUpstreamHandler {

		@Override
		public void messageReceived(ChannelHandlerContext ctx, MessageEvent e)
				throws Exception {
				
			HeartbeatInvalidResponseMessage msg = (HeartbeatInvalidResponseMessage)e.getMessage();
			log.debug("消息失效响应消息处理器,消息:{}", msg);
			Integer key = Integer.valueOf(msg.getFrom().hashCode());
			EndpointMeta item = openMetas.get(key);
			if (null != item && item.heartbeatState.getPulseVersion() <= msg.getPulseVersion()
					&& item.heartbeatState.getPulseTime() <= msg.getPulseTime()) { 
				
				item.memberState = MemberState.OPEN;
				item.heartbeatState.setPulseVersion(msg.getPulseVersion());
				item.heartbeatState.setPulseTime(msg.getPulseTime());
				item.heartbeatState.setLifecycle(msg.getLifecycle());
				item.heartbeatState.setApperceiveTime(System.currentTimeMillis());
			} else {
				if (null == item) {
					log.error("未找到端点[{}]", msg.getFrom());
				} else {
					log.error("心跳失效响应消息无效, 原心跳[{}], 消息[{}]", item.heartbeatState, msg);
				}
			}
		}
	}
	
	/**
	 * 端点成员关闭消息
	 * @author yuyoo (yuyoo4j@163.com)
	 * @date 2009-12-7 上午10:14:06
	 */
	class MemberOpenToCloseMessageHandler  extends SimpleChannelUpstreamHandler {
		
		@Override
		public void messageReceived(ChannelHandlerContext ctx, MessageEvent e)
				throws Exception {
			
			MemberOpenToCloseMessage msg = (MemberOpenToCloseMessage)e.getMessage();
			log.debug("端点成员关闭消息.消息:{}", msg);
			Integer key = Integer.valueOf(msg.getNode().hashCode());
			final EndpointMeta item = openMetas.get(key);
			if (null != item ) {
				
				if (item.heartbeatState.getPulseVersion() <= msg.getPulseVersion()
						&& item.heartbeatState.getPulseTime() <= msg.getPulseTime()) {
					item.memberState = MemberState.CLOSE;
					openMetas.remove(key);
					closeMetas.put(key, item); 
					// 通知 观察者
					noticeMemberStateObserver(MemberState.CLOSE, item);  
				}				
				if (msg.getHopToLive() > 0) {
					msg.setHopToLive(msg.getHopToLive() - 1);
					gossipToNext(msg);
				}
			} else {
				log.debug("收到端点成员关闭消息, msg:{}", msg);
			}
		}
	}
	
	/**
	 * 关闭转开放状态消息处理器
	 * @author yuyoo (yuyoo4j@163.com)
	 * @date 2009-12-7 下午12:06:07
	 */
	class MemberCloseToOpenMessageHandler  extends SimpleChannelUpstreamHandler {
		
		@Override
		public void messageReceived(ChannelHandlerContext ctx, MessageEvent e)
				throws Exception { 
			
			MemberCloseToOpenMessage msg = (MemberCloseToOpenMessage)e.getMessage();
			log.debug("关闭转开放状态消息处理器,消息:{}", msg);
			
			Integer key = Integer.valueOf(msg.getNode().hashCode());
			final EndpointMeta item = closeMetas.get(key);
			if (null != item) {
				
				if (item.heartbeatState.getPulseVersion() < msg.getPulseVersion()
						&& item.heartbeatState.getPulseTime() < msg.getPulseTime()) {
					item.heartbeatState.setPulseVersion(msg.getPulseVersion());
					item.heartbeatState.setPulseTime(msg.getPulseTime());
					item.heartbeatState.setLifecycle(msg.getLifecycle());
					item.heartbeatState.setApperceiveTime(System.currentTimeMillis());
					openMetas.put(key, item);
					// 通知 观察者
					noticeMemberStateObserver(MemberState.OPEN, item);  
				}
				if (msg.getHopToLive() > 0) {
					msg.setHopToLive(msg.getHopToLive() - 1);
					gossipToNext(msg);
				}
			} else {
				log.debug("在关闭状态端点集合中未找到端点. msg[{}]", msg);
			}
			
		}
	}
	
	/**
	 * 端点加入消息处理器
	 * @author yuyoo (yuyoo4j@163.com)
	 * @date 2009-12-7 下午01:11:55
	 */
	class MemberJoinMessageHandler extends SimpleChannelUpstreamHandler {
		
		@Override
		public void messageReceived(ChannelHandlerContext ctx, MessageEvent e)
				throws Exception {
		 
			MemberJoinMessage msg = (MemberJoinMessage)e.getMessage();
			log.debug("端点加入消息处理器,消息:{}", msg);
			
			Integer key = Integer.valueOf(msg.getNode().hashCode());
			EndpointMeta item = joinMetas.get(key);
			if (null == item) {
				
				log.debug("joinMetas中没有找到端点,构造新端点");
			
				item = new EndpointMeta();
				item.endpoint = msg.getNode();
				item.heartbeatState = new HeartBeatState();
				item.heartbeatState.setPulseVersion(msg.getPulseVersion());
				item.heartbeatState.setPulseTime(msg.getPulseTime());
				item.heartbeatState.setLifecycle(msg.getLifecycle());
				item.heartbeatState.setApperceiveTime(System.currentTimeMillis());
				item.memberState = MemberState.JOIN;
				joinMetas.put(key, item);
				// 痛着 观察者
				noticeMemberStateObserver(MemberState.JOIN, item); 
				
				if (msg.getHopToLive() > 0) {
					msg.setHopToLive(msg.getHopToLive() - 1);
					gossipToNext(msg);
				}
			} else if (item.heartbeatState.getPulseVersion() < msg.getPulseVersion()
							&& item.heartbeatState.getPulseTime() < msg.getPulseTime()){
				 
				log.debug("joinMetas中已存在端点,更新心跳信息.");
				item.heartbeatState.setPulseVersion(msg.getPulseVersion());
				item.heartbeatState.setPulseTime(msg.getPulseTime());
				item.heartbeatState.setLifecycle(msg.getLifecycle());
				item.heartbeatState.setApperceiveTime(System.currentTimeMillis());
			}
		}
	}

	/**
	 * 加入转开发状态消息处理器
	 * @author yuyoo (yuyoo4j@163.com)
	 * @date 2009-12-7 下午01:35:08
	 */
	class MemberJoinToOpenMessageHandler extends SimpleChannelUpstreamHandler {
		
		@Override
		public void messageReceived(ChannelHandlerContext ctx, MessageEvent e)
				throws Exception { 
			
			MemberJoinToOpenMessage msg = (MemberJoinToOpenMessage)e.getMessage();
			log.debug("加入转开发状态消息处理器,消息:{}", msg);
			
			Integer key = Integer.valueOf(msg.getNode().hashCode());
			final EndpointMeta item = joinMetas.get(key);
			if (null != item) {
				
				if (item.heartbeatState.getPulseVersion() < msg.getPulseVersion()
						&& item.heartbeatState.getPulseTime() < msg.getPulseTime()) {
					item.memberState = MemberState.OPEN;
					joinMetas.remove(key);
					item.heartbeatState.setPulseVersion(msg.getPulseVersion());
					item.heartbeatState.setPulseTime(msg.getPulseTime());
					item.heartbeatState.setLifecycle(msg.getLifecycle());
					item.heartbeatState.setApperceiveTime(System.currentTimeMillis());
					openMetas.put(key, item);
					// 通知 观察者
					noticeMemberStateObserver(MemberState.OPEN, item);  
				}
				if (msg.getHopToLive() > 0) {
					msg.setHopToLive(msg.getHopToLive() - 1);
					gossipToNext(msg);
				}
			} else {
				log.debug("在加入状态端点集合未找到端点.");
			}
		}
	}
	
	/**
	 * 端点请求消息处理器
	 * @author yuyoo (yuyoo4j@163.com)
	 * @date Dec 6, 2009 12:43:25 PM
	 */
	class MemberSetRequestMessageHandler extends SimpleChannelUpstreamHandler {

		@Override
		public void messageReceived(ChannelHandlerContext ctx, MessageEvent e)
				throws Exception {
			
			MemberSetRequestMessage msg = (MemberSetRequestMessage)e.getMessage();
			log.debug("端点请求消息处理器,消息:{}", msg);
			
			int rc = msg.getRequestMemberStateCode();
			MemberSetResponseMessage res = new MemberSetResponseMessage();
			res.setFrom(currentMeta.endpoint);
			
			List<Endpoint> oe = new ArrayList<Endpoint>();
			if ((rc & MemberState.OPEN.getCode()) == MemberState.OPEN.getCode()) {
				 for (EndpointMeta item : openMetas.values()) {
					 oe.add(item.endpoint);
				 }
			}
			res.setOpenEndpoints(oe);
			
			List<Endpoint> ce = new ArrayList<Endpoint>();
			if ((rc & MemberState.CLOSE.getCode()) == MemberState.CLOSE.getCode()) {
				for (EndpointMeta item : closeMetas.values()) {
					ce.add(item.endpoint);
				}
			}
			res.setCloseEndpoints(ce);
			
			List<Endpoint> je = new ArrayList<Endpoint>();
			if ((rc & MemberState.JOIN.getCode()) == MemberState.JOIN.getCode()) {
				for (EndpointMeta item : joinMetas.values()) {
					je.add(item.endpoint);
				}
			}
			res.setJoinEndpoints(je); 
			
			sendMessage(msg.getFrom(), res);
		}
	}
	
	/**
	 * 端点集合响应消息处理器
	 * @author yuyoo (yuyoo4j@163.com)
	 * @date Dec 6, 2009 12:55:37 PM
	 */
	class MemberSetResponseMessageHandler extends SimpleChannelUpstreamHandler {

		@Override
		public void messageReceived(ChannelHandlerContext ctx, MessageEvent e)
				throws Exception {
			
			MemberSetResponseMessage msg = (MemberSetResponseMessage)e.getMessage();
			log.debug("端点集合响应消息处理器,消息:{}", msg);
			
			List<Endpoint> oe = msg.getOpenEndpoints();
			for (Endpoint item : oe) {
				Integer key = Integer.valueOf(item.hashCode());
				EndpointMeta node = new EndpointMeta();
				node.endpoint = item;
				node.heartbeatState = new HeartBeatState();
				node.memberState = MemberState.OPEN;
				openMetas.put(key, node);				
			}
			
			List<Endpoint> ce = msg.getCloseEndpoints();
			for (Endpoint item : ce) {
				Integer key = Integer.valueOf(item.hashCode());
				EndpointMeta node = new EndpointMeta();
				node.endpoint = item;
				node.heartbeatState = new HeartBeatState();
				node.memberState = MemberState.CLOSE;
				closeMetas.put(key, node);
			}
			
			List<Endpoint> je = msg.getJoinEndpoints();
			for (Endpoint item : je) {
				Integer key = Integer.valueOf(item.hashCode());
				EndpointMeta node = new EndpointMeta();
				node.endpoint = item;
				node.heartbeatState = new HeartBeatState();
				node.memberState = MemberState.JOIN;
				joinMetas.put(key, node);
			} 
		}
	}
}
